package com.ruoyi.quartz.task;

import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.agent.domain.KwMachineAgent;
import com.ruoyi.agent.service.IKwMachineAgentService;
import com.ruoyi.attachments.domain.KwPackageInfo;
import com.ruoyi.attachments.service.IKwPackageInfoService;
import com.ruoyi.cluster.domain.KwClusterInfo;
import com.ruoyi.cluster.domain.KwClusterNodeInfo;
import com.ruoyi.cluster.domain.KwHugeClusterInfo;
import com.ruoyi.cluster.service.IKwClusterInfoService;
import com.ruoyi.cluster.service.IKwClusterNodeInfoService;
import com.ruoyi.cluster.service.IKwHugeClusterInfoService;
import com.ruoyi.common.config.RuoYiConfig;
import com.ruoyi.common.constant.ClusterConstant;
import com.ruoyi.common.constant.HostConstant;
import com.ruoyi.common.constant.LogConstant;
import com.ruoyi.common.constant.TaskConstant;
import com.ruoyi.common.exception.ServiceException;
import com.ruoyi.common.utils.ClusterNode;
import com.ruoyi.common.utils.DiskUtils;
import com.ruoyi.common.utils.LogUtils;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.common.utils.agent.AgentRequest;
import com.ruoyi.common.utils.agent.AgentRestUtil;
import com.ruoyi.common.utils.agent.AgentResult;
import com.ruoyi.common.utils.jsch.JschAuth;
import com.ruoyi.common.utils.jsch.JschResult;
import com.ruoyi.common.utils.jsch.JschUtil;
import com.ruoyi.common.utils.kingwow.ConnParams;
import com.ruoyi.common.utils.kingwow.KingwowUtils;
import com.ruoyi.common.utils.uuid.UUID;
import com.ruoyi.credential.domain.KwCredentialAccess;
import com.ruoyi.credential.domain.KwUserCredential;
import com.ruoyi.credential.service.IKwCredentialAccessService;
import com.ruoyi.credential.service.IKwUserCredentialService;
import com.ruoyi.machine.domain.KwMachineInfo;
import com.ruoyi.machine.service.IKwMachineInfoService;
import com.ruoyi.task.domain.KwTaskInfo;
import com.ruoyi.task.domain.KwTaskLog;
import com.ruoyi.task.service.IKwTaskInfoService;
import com.ruoyi.task.service.IKwTaskLogService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.security.core.parameters.P;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.io.File;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

/**
 * 集群管理
 * 
 * @author gonianer
 */
@Component("clusterTaskHandler")
public class ClusterTaskHandler
{
    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private IKwHugeClusterInfoService clusterService;
    @Autowired
    private IKwClusterInfoService subClusterService;
    @Autowired
    private IKwClusterNodeInfoService nodeService;
    @Autowired
    private IKwMachineInfoService machineService;
    @Autowired
    private IKwMachineAgentService agentService;
    @Autowired
    private IKwPackageInfoService packageServie;
    @Autowired
    private IKwTaskInfoService taskInfoService;
    @Autowired
    private IKwTaskLogService taskLogService;

    @Async
    public void addCluster(KwTaskInfo t,boolean isRetry){
        this.updateTaskStatus(t.getId(),TaskConstant.STATUS_RUNNING,isRetry);
        try {
            KwTaskInfo query = new KwTaskInfo();
            query.setpId(t.getId());
            List<KwTaskInfo> tasks = taskInfoService.selectKwTaskInfoList(query);
            KwHugeClusterInfo cluster = clusterService.selectKwHugeClusterInfoById(t.getJobId());
            if(cluster == null){
                throw new ServiceException("集群信息：id="+t.getJobId()+"不存在");
            }
            KwClusterNodeInfo nodeQuery = new KwClusterNodeInfo();
            nodeQuery.setHclusterId(cluster.getHclusterId());
            List<KwClusterNodeInfo> nodes = nodeService.selectKwClusterNodeInfoList(nodeQuery);
            if(CollectionUtils.isEmpty(nodes)){
                throw new ServiceException("集群：id="+t.getJobId()+"，节点信息不存在");
            }
            Map<String,List<KwClusterNodeInfo>> nodemap = new HashMap<>();
            for(KwClusterNodeInfo node : nodes){
                if(nodemap.get(node.getHostIp()) == null){
                    KwMachineInfo machine = machineService.selectKwMachineInfoByIp(node.getHostIp());
                    KwMachineAgent agent = agentService.selectKwMachineAgent(machine.getId());
                    KwClusterInfo subCluster = subClusterService.selectKwClusterInfoById(node.getClusterId().intValue());
                    node.setAgent(agent);
                    node.setClusterInnerId(Integer.valueOf(subCluster.getInnerId()));
                    List<KwClusterNodeInfo> list = new ArrayList<>();
                    list.add(node);
                    nodemap.put(node.getHostIp(),list);
                }else{
                    node.setAgent(nodemap.get(node.getHostIp()).get(0).getAgent());
                    node.setClusterInnerId(nodemap.get(node.getHostIp()).get(0).getClusterInnerId());
                    nodemap.get(node.getHostIp()).add(node);
                }
            }
            boolean flag = true;
            boolean retry = false;
            for(KwTaskInfo task : tasks){
                try {
                    if(isRetry && TaskConstant.STATUS_FAILED == task.getTaskStat()){
                        retry = true;
                    }
                    if(retry || !isRetry) {
                        this.updateTaskStatus(task.getId(), TaskConstant.STATUS_RUNNING,isRetry);
                        if ("checkClusterCreate".equals(task.getTaskType()))
                            this.checkClusterCreate(task,nodemap);
                        if ("uploadDbResource".equals(task.getTaskType()))
                            this.uploadDbResource(task,cluster,nodemap);
                        if ("decompressAndCheck".equals(task.getTaskType()))
                            this.decompressAndCheck(task,cluster,nodemap);
                        if ("linkDataDir".equals(task.getTaskType()))
                            this.linkDataDir(task,cluster,nodemap);
                        if ("startProcess".equals(task.getTaskType())) {
                            if (retry) {
                                this.rollbackStartProcess(task, cluster, nodemap);
                            }
                            this.startProcess(task, cluster, nodemap);
                        }
                        if ("checkProcessStatus".equals(task.getTaskType()))
                            this.checkProcessStatus(task,cluster,nodemap);
                        if ("bootstrapCluster".equals(task.getTaskType()))
                            this.bootstrapCluster(task,cluster,nodemap);
                        if ("waitForClusterOk".equals(task.getTaskType()))
                            this.waitForClusterOk(task,cluster,nodemap);
                        if ("updateClusterStatus".equals(task.getTaskType()))
                            this.updateClusterStatus(task,cluster,nodemap);

                        this.updateTaskStatus(task.getId(), TaskConstant.STATUS_SUCCESS,isRetry);
                    }
                } catch (Exception e) {
                    logger.error("执行任务失败:" + e.getMessage());
                    this.recordLog(t.getpId(),t.getId(),LogConstant.ERROR,
                            "执行任务失败："+ e.getMessage());
                    this.updateTaskStatus(task.getId(), TaskConstant.STATUS_FAILED,isRetry);
                    this.updateTaskStatus(t.getId(), TaskConstant.STATUS_FAILED,isRetry);
                    flag = false;
                    break;
                }
            };
            if(flag) {
                this.updateTaskStatus(t.getId(), TaskConstant.STATUS_SUCCESS,isRetry);
                this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,
                        "创建集群任务完成：status="+ ClusterConstant.STATUS_RUNNING);
            }
        }catch(Exception e){
            logger.error(e.getMessage());
            this.recordLog(t.getpId(),t.getId(), LogConstant.ERROR,e.getMessage());
            this.updateTaskStatus(t.getId(), TaskConstant.STATUS_FAILED,isRetry);
        }
    }

    @Async
    public void rollbackCluster(Long taskId){
        boolean retry = false;
        KwTaskInfo t = taskInfoService.selectKwTaskInfoById(taskId.intValue());
        this.updateTaskStatus(t.getId(),TaskConstant.STATUS_RUNNING,retry);
        try {
            KwTaskInfo query = new KwTaskInfo();
            query.setpId(t.getId());
            List<KwTaskInfo> tasks = taskInfoService.selectKwTaskInfoList(query);
            Collections.reverse(tasks);

            KwHugeClusterInfo cluster = clusterService.selectKwHugeClusterInfoById(t.getJobId());
            if(cluster == null){
                throw new ServiceException("集群信息：id="+t.getJobId()+"不存在");
            }
            KwClusterNodeInfo nodeQuery = new KwClusterNodeInfo();
            nodeQuery.setHclusterId(cluster.getHclusterId());
            List<KwClusterNodeInfo> nodes = nodeService.selectKwClusterNodeInfoList(nodeQuery);
            if(CollectionUtils.isEmpty(nodes)){
                throw new ServiceException("集群：id="+t.getJobId()+"，节点信息不存在");
            }
            Map<String,List<KwClusterNodeInfo>> nodemap = new HashMap<>();
            for(KwClusterNodeInfo node : nodes){
                if(nodemap.get(node.getHostIp()) == null){
                    KwMachineInfo machine = machineService.selectKwMachineInfoByIp(node.getHostIp());
                    KwMachineAgent agent = agentService.selectKwMachineAgent(machine.getId());
                    KwClusterInfo subCluster = subClusterService.selectKwClusterInfoById(node.getClusterId().intValue());
                    node.setAgent(agent);
                    node.setClusterInnerId(Integer.valueOf(subCluster.getInnerId()));
                    List<KwClusterNodeInfo> list = new ArrayList<>();
                    list.add(node);
                    nodemap.put(node.getHostIp(),list);
                }else{
                    node.setAgent(nodemap.get(node.getHostIp()).get(0).getAgent());
                    node.setClusterInnerId(nodemap.get(node.getHostIp()).get(0).getClusterInnerId());
                    nodemap.get(node.getHostIp()).add(node);
                }
            }
            boolean flag = false;
            for(KwTaskInfo task : tasks){
                try {
                    if(TaskConstant.STATUS_FAILED == task.getTaskStat()) {
                        flag = true;
                    }
                    if(flag) {
                        this.updateTaskStatus(task.getId(), TaskConstant.STATUS_RUNNING,retry);
                        if ("checkClusterCreate".equals(task.getTaskType()))
                            this.rollbackCheckClusterCreate(task,nodemap);
                        if ("uploadDbResource".equals(task.getTaskType()))
                            this.rollbackUploadDbResource(task,cluster,nodemap);
                        if ("decompressAndCheck".equals(task.getTaskType()))
                            this.rollbackDecompressAndCheck(task,cluster,nodemap);
                        if ("linkDataDir".equals(task.getTaskType()))
                            this.rollbackLinkDataDir(task,cluster,nodemap);
                        if ("startProcess".equals(task.getTaskType()))
                            this.rollbackStartProcess(task,cluster,nodemap);
                        if ("checkProcessStatus".equals(task.getTaskType()))
                            this.rollbackCheckProcessStatus(task,cluster,nodemap);
                        if ("bootstrapCluster".equals(task.getTaskType()))
                            this.rollbackBootstrapCluster(task,cluster,nodemap);
                        if ("waitForClusterOk".equals(task.getTaskType()))
                            this.rollbackWaitForClusterOk(task,cluster,nodemap);
                        if ("updateClusterStatus".equals(task.getTaskType()))
                            this.rollbackUpdateClusterStatus(task,cluster,nodemap);

                        this.updateTaskStatus(task.getId(), TaskConstant.STATUS_WAIT,retry);
                    }
                } catch (Exception e) {
                    logger.error("执行任务失败:" + e.getMessage());
                    this.updateTaskStatus(task.getId(), TaskConstant.STATUS_FAILED,retry);
                    this.updateTaskStatus(t.getId(), TaskConstant.STATUS_FAILED,retry);
                    flag = false;
                    break;
                }
            };

            //删除节点
            nodeService.deleteKwClusterNodeInfoByHclusterId(cluster.getHclusterId());
            //删除子集群
            subClusterService.deleteKwClusterInfoByHclusterId(cluster.getHclusterId());
            //删除大集群
            clusterService.deleteKwHugeClusterInfoById(cluster.getId());

            this.updateTaskStatus(t.getId(), TaskConstant.STATUS_SUCCESS,retry);
            this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"放弃添加集群任务完成");
        }catch(Exception e){
            logger.error(e.getMessage());
            this.recordLog(t.getpId(),t.getId(), LogConstant.ERROR,e.getMessage());
            this.updateTaskStatus(t.getId(), TaskConstant.STATUS_SUCCESS,retry);
        }
    }

    @Async
    public void retryAddCluster(Long taskId){
        KwTaskInfo t = taskInfoService.selectKwTaskInfoById(taskId.intValue());
        try {
            if(t == null){
                throw new ServiceException("任务新：id="+taskId+"不存在");
            }
            this.addCluster(t,true);
        }catch(Exception e){
            logger.error(e.getMessage());
            this.recordLog(t.getpId(),t.getId(), LogConstant.ERROR,e.getMessage());
            this.updateTaskStatus(t.getId(), TaskConstant.STATUS_FAILED,true);
        }
    }

    /**
     * 更新任务状态
     * @param taskId
     * @param status
     */
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public void updateTaskStatus(Long taskId,Integer status,boolean retry){
        KwTaskInfo taskInfo = new KwTaskInfo();
        taskInfo.setId(taskId);
        taskInfo.setTaskStat(status);
        if(status.intValue() == TaskConstant.STATUS_RUNNING) {
            if(!retry)
                taskInfo.setStartTime(new Date());
            taskInfo.setEndTime(null);
        }else {
            taskInfo.setEndTime(new Date());
        }
        taskInfoService.updateKwTaskInfo(taskInfo);
    }

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public void recordLog(Long pId,Long taskId,String level,String msg){
        KwTaskLog log = new KwTaskLog();
        log.setTaskId(taskId);
        log.setpId(pId);
        log.setLogLevel(level);
        msg = LogUtils.getLog(msg);
        log.setLogMsg(msg);
        taskLogService.insertKwTaskLog(log);
    }

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean checkClusterCreate(KwTaskInfo t,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"开始执行预检查任务");
        nodes.keySet().forEach(hostIp->{
            this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"检查机器:"+hostIp+"，jwagent状态");
            KwMachineAgent agent = nodes.get(hostIp).get(0).getAgent();
            AgentRequest request = new AgentRequest(hostIp,agent.getPort(),agent.getHttpUser(),agent.getHttpPasswd());
            request.setCommand("date");
            this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,
                    "post "+hostIp+" for date exec ");
            AgentResult result = AgentRestUtil.postAgentCommand(request);
            this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,
                    "post "+hostIp+" for date exec result: retcode:" +result.getData().getRetcode()
                            +",stdout:"+result.getData().getStdout()
                            +",stderr:"+result.getData().getStderr()
            );
            if(result.getData().getRetcode()!=0){
                throw new ServiceException("机器:"+hostIp+"，jwagent状态异常");
            }
        });

        return flag;
    }
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean rollbackCheckClusterCreate(KwTaskInfo t,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"任务回滚，无需操作");
        return flag;
    }

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean uploadDbResource(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"开始准备上传资源");
        KwPackageInfo packageQuery = new KwPackageInfo();
        packageQuery.setVersion(cluster.getVersion());
        KwPackageInfo filepackage = packageServie.selectKwPackageInfoById(Long.valueOf(cluster.getVersion()));
        if(filepackage==null){
            throw new ServiceException("数据库版本："+cluster.getVersion()+"对应资源包不存在");
        }
        String filePath = RuoYiConfig.getProfile()+filepackage.getPackagePath();
        File file = new File(filePath);
        if(!file.exists()){
            throw new ServiceException("资源文件："+filePath+"不存在");
        }
        nodes.keySet().forEach(hostIp->{
            CompletableFuture<String> future = CompletableFuture.supplyAsync(()-> {
                //检查目标文件是否存在
                String fileName = filePath.substring(filePath.lastIndexOf("/")+1);
                KwMachineAgent agent = nodes.get(hostIp).get(0).getAgent();
                AgentRequest request = new AgentRequest(hostIp, agent.getPort(), agent.getHttpUser(), agent.getHttpPasswd());
                request.setCommand(fileName);
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                        "检查文件 " + hostIp + " 是否存在，文件名: "+fileName);
                AgentResult result = AgentRestUtil.getHostFileExist(request);
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                        "检查文件 " + hostIp + " ，结果: "+JSONObject.toJSONString(result.getData()));
                if(result.getData().getExisted()){
                    this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                            "检查文件 " + hostIp + " ，文件名: "+fileName+"已存在，无需上传");
                    return result.getData().getFilepath();
                }
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                        "post " + hostIp + " for upload file: "+filePath);

                request.setCommand(filePath);
                result = AgentRestUtil.postAgentUpload(request);
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                        "上传资源： " + hostIp + " 结束，上传结果："+ JSONObject.toJSONString(result.getData()));
                if(!result.getData().isSuccess()){
                    throw new ServiceException("上传资源： " + hostIp + " 结束，上传失败。");
                }
                return result.getData().getLocalpath();
            });
            future.join();
        });

        return flag;
    }
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean rollbackUploadDbResource(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"开始准备清理数据库安装包");
        KwPackageInfo packageQuery = new KwPackageInfo();
        packageQuery.setVersion(cluster.getVersion());
        KwPackageInfo filepackage = packageServie.selectKwPackageInfoById(Long.valueOf(cluster.getVersion()));
        if(filepackage==null){
            throw new ServiceException("数据库版本："+cluster.getVersion()+"对应资源包不存在");
        }
        String filePath = filepackage.getPackagePath();
        String fileName = filePath.substring(filePath.lastIndexOf("/")+1);
        nodes.keySet().forEach(hostIp->{
            KwMachineAgent agent = nodes.get(hostIp).get(0).getAgent();
            AgentRequest request = new AgentRequest(hostIp, agent.getPort(), agent.getHttpUser(), agent.getHttpPasswd());
            StringBuffer command = new StringBuffer(AgentRestUtil.COMMAND_RM_RF_DATA_DIR
                    .replaceAll("data_dir","~/"+agent.getInstallHome()+"/upload/"+fileName));
            request.setCommand(command.toString());
            this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                    "准备清理数据库安装包： " + hostIp + " 执行命令："+ command);
            AgentResult result = AgentRestUtil.postAgentCommand(request);
            this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                    "清理数据库安装包： " + hostIp + " 处理结果："+ JSONObject.toJSONString(result.getData()));
            if(result.getData().getRetcode()!=0){
                throw new ServiceException("清理数据库安装包： " + hostIp + " 失败："+result.getData().getStderr());
            }
        });
        return flag;
    }

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean decompressAndCheck(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"开始准备解压资源文件");
        KwPackageInfo packageQuery = new KwPackageInfo();
        packageQuery.setVersion(cluster.getVersion());
        KwPackageInfo filepackage = packageServie.selectKwPackageInfoById(Long.valueOf(cluster.getVersion()));
        if(filepackage==null){
            throw new ServiceException("数据库版本："+cluster.getVersion()+"对应资源包不存在");
        }
        String filePath = filepackage.getPackagePath();
        String fileName = filePath.substring(filePath.lastIndexOf("/")+1);
        nodes.keySet().forEach(hostIp->{
            CompletableFuture<String> future = CompletableFuture.supplyAsync(()-> {
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                        "post " + hostIp + " for decompress file: "+fileName +"to install home:"+cluster.getKwRootDir());
                KwMachineAgent agent = nodes.get(hostIp).get(0).getAgent();
                AgentRequest request = new AgentRequest(hostIp, agent.getPort(), agent.getHttpUser(), agent.getHttpPasswd());
                String command = AgentRestUtil.POST_DECOMPRESS_RPM.replaceAll("install_dir","~/"+cluster.getKwRootDir())
                        .replaceAll("rmp_filepath","~/"+agent.getInstallHome()+"/upload/"+fileName);
                request.setCommand(command);
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                        "资源解压结束： " + hostIp + " 执行命令："+ command);
                AgentResult result = AgentRestUtil.postAgentCommand(request);
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                        "资源解压结束： " + hostIp + " 处理结果："+ JSONObject.toJSONString(result.getData()));
                if(result.getData().getRetcode()!=0){
                    throw new ServiceException("资源解压： " + hostIp + " 结束，解压失败。");
                }
                return result.getData().getStderr();
            });
            future.join();
        });

        return flag;
    }
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean rollbackDecompressAndCheck(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"开始准备清理数据库安装目录");

        nodes.keySet().forEach(hostIp->{
            KwMachineAgent agent = nodes.get(hostIp).get(0).getAgent();
            AgentRequest request = new AgentRequest(hostIp, agent.getPort(), agent.getHttpUser(), agent.getHttpPasswd());
            StringBuffer command = new StringBuffer(AgentRestUtil.COMMAND_RM_RF_DATA_DIR
                    .replaceAll("data_dir","~/"+cluster.getKwRootDir()+"/*"));
            request.setCommand(command.toString());
            this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                    "准备清理数据库安装目录： " + hostIp + " 执行命令："+ command);
            AgentResult result = AgentRestUtil.postAgentCommand(request);
            this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                    "清理数据库安装目录： " + hostIp + " 处理结果："+ JSONObject.toJSONString(result.getData()));
            if(result.getData().getRetcode()!=0){
                throw new ServiceException("清理数据库安装目录： " + hostIp + " 失败："+result.getData().getStderr());
            }
        });

        return flag;
    }

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean linkDataDir(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"开始准备建立数据目录连接");

        Map<String,Object> extra = JSONObject.parseObject(cluster.getExtraInfo(),Map.class);
        int dataMode = (Integer)extra.get("dataMode");
        String dataPath = (String)extra.get("dataPath");

        nodes.keySet().forEach(hostIp->{
            CompletableFuture<String> future = CompletableFuture.supplyAsync(()-> {
                KwMachineAgent agent = nodes.get(hostIp).get(0).getAgent();
                AgentRequest request = new AgentRequest(hostIp, agent.getPort(), agent.getHttpUser(), agent.getHttpPasswd());
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                        "post " + hostIp + " for rebuilddir ,dataMode is : "+dataMode);
                // dataMode == 0 需要自动创建数据子目录
                List<String> dgDataList = (List)extra.get("dgData");
                if(dgDataList==null) dgDataList=new ArrayList<>();
                String dgData = String.join(",", dgDataList);
                String tgData = (String)extra.get("tgData");
                String tgLog = (String)extra.get("tgLog");
                String cgroupDir = (String)extra.get("cgroupDir");
                if(dataMode == 0){
                    request.setCommand(AgentRestUtil.POST_GET_DIR_DISK.replaceAll("data_path",dataPath));
                    AgentResult result = AgentRestUtil.postAgentCommand(request);
                    this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                            "获取磁盘大小： " + hostIp + " 处理结果："+ JSONObject.toJSONString(result.getData()));
                    if(result.getData().getRetcode()!=0){
                        throw new ServiceException("获取磁盘大小： " + hostIp + " 结束，建立数据目录连接失败。");
                    }else{
                        int dataSize = Integer.valueOf(result.getData().getStdout().trim());
                        int dirCount = DiskUtils.calcDirCount(dataSize);
                        List<String> dataList = new ArrayList<>();
                        for(int i=1;i<=dirCount;i++){
                            dataList.add(dataPath+"/"+i);
                        }
                        dgData = String.join(",", dataList);
                        tgData = dataPath+"/tg_data";
                        tgLog = dataPath+"/tg_commit_log";
                        String command = AgentRestUtil.POST_MK_DIR_DISK
                                .replaceAll("dirlist",dgData+","+tgData+","+tgLog);
                        this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                                "自动创建目录： " + hostIp + " 执行命令："+ command);
                        request.setCommand(command);
                        result = AgentRestUtil.postAgentCommand(request);
                        this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                                "自动创建目录： " + hostIp + " 处理结果："+ JSONObject.toJSONString(result.getData()));
                        if(result.getData().getRetcode()!=0){
                            throw new ServiceException("自动创建目录： " + hostIp + " 失败，建立数据目录连接失败。");
                        }
                        //更新磁盘目录
                        extra.put("dgData",dataList);
                        extra.put("tgData",tgData);
                        extra.put("tgLog",tgLog);
                        KwHugeClusterInfo extraUpdate = new KwHugeClusterInfo();
                        extraUpdate.setId(cluster.getId());
                        extraUpdate.setExtraInfo(JSONObject.toJSONString(extra));
                        clusterService.updateKwHugeClusterInfo(extraUpdate);
                    }
                }
                StringBuffer command = new StringBuffer(AgentRestUtil.POST_REBULID_DIR.
                        replaceAll("scripts_dir","~/"+agent.getInstallHome()+"/scripts"));
                command.append(" -s ~/"+cluster.getKwRootDir());
                command.append(" -m independent_mapping_dir");
                command.append(" -d "+cluster.getHclusterId());
                command.append(" -l "+cluster.getHclusterId());
                command.append(" -a "+dgData);
                command.append(" -b "+tgData);
                command.append(" -f "+tgLog);
                command.append(" -g "+(cgroupDir==null?"''":cgroupDir));
                nodes.get(hostIp).stream().forEach(node->{
                    if(!"cg".equals(node.getNodeType())){
                        command.append(" -r "+node.getNodeType());
                        this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                                "建立数据目录连接： " + hostIp + " 执行命令："+ command.toString());
                        request.setCommand(command.toString());
                        AgentResult result = AgentRestUtil.postAgentCommand(request);
                        this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                                "建立数据目录连接： " + hostIp + " 处理结果："+ JSONObject.toJSONString(result.getData()));
                        if(result.getData().getRetcode()!=0){
                            throw new ServiceException("建立数据目录连接： " + hostIp + " 失败："+result.getData().getStderr());
                        }
                    }
                });
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,"建立数据目录连接： " + hostIp + " 完成");
                return "success";
            });
            future.join();
        });

        return flag;
    }
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean rollbackLinkDataDir(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"开始准备清理数据目录");

        Map<String,Object> extra = JSONObject.parseObject(cluster.getExtraInfo(),Map.class);
        nodes.keySet().forEach(hostIp->{
            CompletableFuture<String> future = CompletableFuture.supplyAsync(()-> {
                KwMachineAgent agent = nodes.get(hostIp).get(0).getAgent();
                AgentRequest request = new AgentRequest(hostIp, agent.getPort(), agent.getHttpUser(), agent.getHttpPasswd());
                List<String> dgData = (List)extra.get("dgData");
                if(dgData==null) dgData = new ArrayList<>();
                String tgData = (String)extra.get("tgData");
                String tgLog = (String)extra.get("tgLog");
                dgData.add(tgData);
                dgData.add(tgLog);

                StringBuffer command = new StringBuffer(AgentRestUtil.COMMAND_RM_RF_DATA_DIR
                        .replaceAll("data_dir",String.join(",",dgData)));
                request.setCommand(command.toString());
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                        "准备清理数据目录： " + hostIp + " 执行命令："+ command);
                AgentResult result = AgentRestUtil.postAgentCommand(request);
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                        "清理数据目录： " + hostIp + " 处理结果："+ JSONObject.toJSONString(result.getData()));
                if(result.getData().getRetcode()!=0){
                    this.recordLog(t.getpId(), t.getId(), LogConstant.WARN,"清理数据目录： " + hostIp + " 失败："+result.getData().getStderr());
                }
                return "success";
            });
            future.join();
        });

        return flag;
    }


    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean startProcess(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"开始准备启动进程");
        int subClusterCnt = subClusterService.selectKwClusterInfoListByHclustId(cluster.getHclusterId());
        int mgCnt = 0;
        int tgCnt = 0;
        TreeMap<Integer,Integer> tgGroupMap = new TreeMap<>();
        for(String hostIp : nodes.keySet()){
            mgCnt+=nodes.get(hostIp).stream().filter(node-> ClusterNode.NODE_MG.equals(node.getNodeType())).count();
            tgCnt+=nodes.get(hostIp).stream().filter(node-> ClusterNode.NODE_TG.equals(node.getNodeType())).count();
            nodes.get(hostIp).stream().forEach(node->{
                if(ClusterNode.NODE_TG.equals(node.getNodeType())){
                    if(tgGroupMap.get(node.getClusterInnerId())==null){
                        tgGroupMap.put(node.getClusterInnerId(),1);
                    }else{
                        tgGroupMap.put(node.getClusterInnerId(),tgGroupMap.get(node.getClusterInnerId())+1);
                    }
                }
            });
        }
        List<Integer> groupList = new ArrayList<>();
        tgGroupMap.keySet().forEach(key ->{
            groupList.add(tgGroupMap.get(key));
        });
        String tgGroupCnt=StringUtils.join(groupList,";");
        int mgTotal = mgCnt;
        int tgTotal = tgCnt;
        nodes.keySet().forEach(hostIp->{
            CompletableFuture<String> future = CompletableFuture.supplyAsync(()-> {
                KwMachineAgent agent = nodes.get(hostIp).get(0).getAgent();
                KwMachineInfo machine = machineService.selectKwMachineInfoByIp(hostIp);

                AgentRequest request = new AgentRequest(hostIp, agent.getPort(), agent.getHttpUser(), agent.getHttpPasswd());
                boolean isMaster = true;
                String masterIp="";
                Integer masterPort=0;
                for(KwClusterNodeInfo node: nodes.get(hostIp)){

                    this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                            "开始准备启动进程IP:" + hostIp +",type:"+node.getNodeType());
                    StringBuffer cmd = new StringBuffer();

                    if(ClusterNode.NODE_MG.equals(node.getNodeType())){
                        cmd.append("bin/manEng");
                        cmd.append(" -r ").append(node.getHostIp()).append(":").append(node.getPort());
                        cmd.append(" -i ").append(machine.getNetCardName());
                        cmd.append(" -C ").append(node.getClusterInnerId());
                        if(isMaster){
                            masterIp = node.getHostIp();
                            masterPort = node.getPort();
                            cmd.append(" -R ").append(node.getHostIp()).append(":").append(node.getPort());
                            cmd.append(" -U ").append(mgTotal);
                            if(cluster.getIsMups()==1){
                                cmd.append(" -u ").append(tgGroupCnt);
                            }else{
                                cmd.append(" -u ").append(tgTotal);
                            }
                        }else{
                            cmd.append(" -R ").append(masterIp).append(":").append(masterPort);
                        }
                        if(cluster.getIsMups()==1){
                            cmd.append(" -K ").append(subClusterCnt);
                            cmd.append(" -G 1");//需要支持多TG
                        }
                    }
                    if(ClusterNode.NODE_TG.equals(node.getNodeType())){
                        cmd.append("bin/transEng");
                        cmd.append(" -r ").append(masterIp).append(":").append(masterPort);
                        cmd.append(" -p ").append(node.getPort());
                        cmd.append(" -m ").append(node.getInnerPort());
                        cmd.append(" -i ").append(machine.getNetCardName());
                        cmd.append(" -C ").append(node.getClusterInnerId());
                        if(cluster.getIsMups()==1){
                            cmd.append(" -g 0");//需要支持多主TG
                        }
                    }
                    if(ClusterNode.NODE_CG.equals(node.getNodeType())){
                        cmd.append("bin/calEng");
                        cmd.append(" -r ").append(masterIp).append(":").append(masterPort);
                        cmd.append(" -p ").append(node.getPort());
                        cmd.append(" -z ").append(node.getInnerPort());
                        cmd.append(" -i ").append(machine.getNetCardName());
                        cmd.append(" -C ").append(node.getClusterInnerId());
                    }
                    if(ClusterNode.NODE_DG.equals(node.getNodeType())){
                        cmd.append("bin/datEng");
                        cmd.append(" -r ").append(masterIp).append(":").append(masterPort);
                        cmd.append(" -p ").append(node.getPort());
                        cmd.append(" -n ").append(cluster.getHclusterId());
                        cmd.append(" -i ").append(machine.getNetCardName());
                        cmd.append(" -C ").append(node.getClusterInnerId());
                    }
                    String command = AgentRestUtil.COMMAND_START_DB_NODE.
                            replaceAll("kw_root_dir",cluster.getKwRootDir())
                            .replaceAll("start_cmd",cmd.toString());
                    request.setCommand(command);
                    AgentResult result = AgentRestUtil.postAgentCommand(request);
                    this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                            "启动节点： " + hostIp + " 类型："+node.getNodeType()+",结果："+ JSONObject.toJSONString(result.getData()));
                    if(result.getData().getRetcode()!=0){
                        throw new ServiceException("启动节点： " + hostIp + " 失败："+result.getData().getStderr());
                    }
                    this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                            "启动节点： " + hostIp + " 类型："+node.getNodeType()+",成功");
                    node.setCmdInfo(command);
                    KwClusterNodeInfo updnode =new KwClusterNodeInfo();
                    updnode.setId(node.getId());
                    updnode.setCmdInfo(command);
                    updnode.setOnlineTime(new Date());
                    updnode.setLatestAliveTime(new Date());
                    if(isMaster){
                        updnode.setNodeRole(ClusterNode.NODE_MASER);
                    }
                    nodeService.updateKwClusterNodeInfo(updnode);


                    isMaster = false;
                }
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,"机器："+hostIp+"，全部节点启动完成");
                return "success";
            });
            future.join();
        });

        try {
            this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,"全部节点启动完成，sleep 30s");
            Thread.sleep(30000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return flag;
    }
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean rollbackStartProcess(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"开始准备结束进程");
        nodes.keySet().forEach(hostIp->{
            CompletableFuture<String> future = CompletableFuture.supplyAsync(()-> {
                KwMachineAgent agent = nodes.get(hostIp).get(0).getAgent();
                AgentRequest request = new AgentRequest(hostIp, agent.getPort(), agent.getHttpUser(), agent.getHttpPasswd());

                for(KwClusterNodeInfo node: nodes.get(hostIp)){
                    String command = AgentRestUtil.COMMAND_KILL_DB_NODE.
                            replaceAll("node_type",ClusterNode.NODE_TYPE_CONVERT.get(node.getNodeType()))
                            .replaceAll("kw_root_dir",cluster.getKwRootDir());
                    request.setCommand(command);
                    this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                            "开始准备结束进程IP:" + hostIp +",type:"+node.getNodeType()+",执行命令："+command);
                    AgentResult result = AgentRestUtil.postAgentCommand(request);
                    this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                            "结束进程节点： " + hostIp + " 类型："+node.getNodeType()+",结果："+ JSONObject.toJSONString(result.getData()));
                    if(result.getData().getRetcode()!=0){
                        throw new ServiceException("结束进程节点： " + hostIp + " 失败："+result.getData().getStderr());
                    }
                    this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                            "结束进程节点： " + hostIp + " 类型："+node.getNodeType()+",成功");
                }
                return "success";
            });
            future.join();
        });
        this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,"全部节点数据库进程结束完成");
        return flag;
    }

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean checkProcessStatus(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"开始检查数据库进程状态");
        nodes.keySet().forEach(hostIp->{
            CompletableFuture<String> future = CompletableFuture.supplyAsync(()-> {
                KwMachineAgent agent = nodes.get(hostIp).get(0).getAgent();
                AgentRequest request = new AgentRequest(hostIp, agent.getPort(), agent.getHttpUser(), agent.getHttpPasswd());

                for(KwClusterNodeInfo node: nodes.get(hostIp)){
                    String command = AgentRestUtil.COMMAND_CHECK_NODE_PROCESS
                            .replaceAll("node_bin",ClusterNode.NODE_TYPE_CONVERT.get(node.getNodeType()))
                            .replaceAll("node_port",String.valueOf(node.getPort()));
                    request.setCommand(command);
                    this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                            "开始检查机器： " + hostIp + " 类型："+node.getNodeType()+"进程状态："+ command);
                    AgentResult result = AgentRestUtil.postAgentCommand(request);
                    this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                            "检查机器： " + hostIp + " 类型："+node.getNodeType()+",结果："+ JSONObject.toJSONString(result.getData()));
                    if(StringUtils.isEmpty(result.getData().getStdout())){
                        throw new ServiceException("机器： " + hostIp + " 类型："+node.getNodeType()+",进程状态异常："+result.getData().getStderr());
                    }
                }
                this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,"机器："+hostIp+"，全部节点启动完成");
                return "success";
            });
            future.join();
        });

        try {
            this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,"全部节点启动完成，sleep 30s");
            Thread.sleep(30);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return flag;
    }
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean rollbackCheckProcessStatus(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"任务回滚，无需操作");
        return flag;
    }

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean bootstrapCluster(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"开始准备初始化集群");
        KwClusterNodeInfo query = new KwClusterNodeInfo();
        query.setNodeRole(ClusterNode.NODE_MASER);
        query.setNodeType(ClusterNode.NODE_MG);
        query.setHclusterId(cluster.getHclusterId());
        List<KwClusterNodeInfo> mgnodes = nodeService.selectKwClusterNodeInfoList(query);
        if(CollectionUtils.isEmpty(mgnodes)){
            throw new ServiceException("集群： " + cluster.getHclusterId() + " 主mg不存在");
        }
        KwClusterNodeInfo masterMg = mgnodes.get(0);
        String hostIp = masterMg.getHostIp();
        KwMachineAgent agent = nodes.get(hostIp).get(0).getAgent();
        AgentRequest request = new AgentRequest(hostIp, agent.getPort(), agent.getHttpUser(), agent.getHttpPasswd());

        String command = AgentRestUtil.COMMAND_BOOTSTRAP_CLUSTER
                .replaceAll("kw_root_dir",cluster.getKwRootDir())
                .replaceAll("master_ip",String.valueOf(masterMg.getHostIp()))
                .replaceAll("master_port",String.valueOf(masterMg.getPort()));
        request.setCommand(command);
        this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                "初始化集群： " + cluster.getHclusterId() + " 执行："+command);
        AgentResult result = AgentRestUtil.postAgentCommand(request);
        this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,
                "初始化集群： " +  cluster.getHclusterId() +",结果："+ JSONObject.toJSONString(result.getData()));
        if(result.getData().getRetcode()!=0){
            throw new ServiceException("初始化集群： " + cluster.getHclusterId() + " 失败："+result.getData().getStderr());
        }
        this.recordLog(t.getpId(), t.getId(), LogConstant.INFO,"初始化集群： " +  cluster.getHclusterId() +",完成");

        return flag;
    }
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean rollbackBootstrapCluster(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"任务回滚，无需操作");
        return flag;
    }

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean waitForClusterOk(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"等待集群状态正常");
        int retryCnt =1;
        boolean isOk = false;
        while(retryCnt<21 && !isOk) {
            for(String hostIp : nodes.keySet()) {
                for (KwClusterNodeInfo node : nodes.get(hostIp)) {
                    if (ClusterNode.NODE_CG.equals(node.getNodeType())) {
                        this.recordLog(t.getpId(), t.getId(), LogConstant.INFO, "第"+retryCnt+"次尝试连接cg:" + node.getHostIp() + ",port:" + node.getInnerPort());
                        try {
                            //连接数据库查询select 1
                            ConnParams params = new ConnParams(node.getHostIp(), node.getInnerPort(), cluster.getAccessKey());
                            KingwowUtils.testConnection(params);
                            this.recordLog(t.getpId(), t.getId(), LogConstant.INFO, "第"+retryCnt+"次连接cg:" + node.getHostIp() + ",port:" + node.getInnerPort() + "，成功");
                            isOk = true;
                        } catch (Exception e) {
                            logger.warn("test connect kingwow db failed ", e.getMessage());
                            this.recordLog(t.getpId(), t.getId(), LogConstant.INFO, "第"+retryCnt+"次连接cg:" + node.getHostIp() + ",port:" + node.getInnerPort() + "，失败：" + e.getMessage());
                            retryCnt++;
                            try {
                                Thread.sleep(5000);
                            } catch (Exception ex) {
                                logger.warn("thread sleep failed", e);
                            }
                        }
                    }
                }
            }
        }
        if(isOk){
            this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"数据库连接测试成功，集群状态正常");
        }else{
            throw new ServiceException("数据库连接测试失败，请检查集群状态");
        }
        return flag;
    }
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean rollbackWaitForClusterOk(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"任务回滚，无需操作");
        return flag;
    }

    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean updateClusterStatus(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"更新集群状态");
        KwHugeClusterInfo hugeCluster = new KwHugeClusterInfo();
        hugeCluster.setId(cluster.getId());
        hugeCluster.setStatus(ClusterConstant.STATUS_RUNNING);
        clusterService.updateKwHugeClusterInfo(hugeCluster);
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"更新集群状态完成：running");
        return flag;
    }
    @Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRES_NEW)
    public boolean rollbackUpdateClusterStatus(KwTaskInfo t,KwHugeClusterInfo cluster,Map<String,List<KwClusterNodeInfo>> nodes){
        boolean flag = false;
        this.recordLog(t.getpId(),t.getId(),LogConstant.INFO,"任务回滚，无需操作");
        return flag;
    }

}
